/**
 * 
 */
package org.perfect.batch.launcher;

import java.time.Duration;

import org.perfect.batch.BatchConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.metrics.BatchMetrics;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;

/**
 * @author billmse
 *
 */
public class ScheduledJobLauncher implements JobLauncher {

	@Autowired
	private JobRepository jobRepository;

	@Autowired
	@Qualifier(BatchConstants.BATCH_TASK_SCHEDULER_NAME)
	private TaskScheduler taskScheduler;

	private Logger logger = LoggerFactory.getLogger(getClass());

	@Override
	public JobExecution run(final Job job, final JobParameters jobParameters)
			throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
			JobParametersInvalidException {

		Assert.notNull(job, "The Job must not be null.");
		Assert.notNull(jobParameters, "The JobParameters must not be null.");

		final JobExecution jobExecution;
		JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
		if (lastExecution != null) {
			if (!job.isRestartable()) {
				throw new JobRestartException("JobInstance already exists and is not restartable");
			}
			/**
			 * validate here if it has stepExecutions that are UNKNOWN, STARTING, STARTED
			 * and STOPPING retrieve the previous execution and check
			 */
			for (StepExecution execution : lastExecution.getStepExecutions()) {
				BatchStatus status = execution.getStatus();
				if (status.isRunning() || status == BatchStatus.STOPPING) {
					throw new JobExecutionAlreadyRunningException(
							"A job execution for this job is already running: " + lastExecution);
				} else if (status == BatchStatus.UNKNOWN) {
					throw new JobRestartException(
							"Cannot restart step [" + execution.getStepName() + "] from UNKNOWN status. "
									+ "The last execution ended with a failure that could not be rolled back, "
									+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
				}
			}
		}

		/**
		 * Check the validity of the parameters before doing creating anything in the
		 * repository...
		 */
		job.getJobParametersValidator().validate(jobParameters);

		/**
		 * There is a very small probability that a non-restartable job can be
		 * restarted, but only if another process or thread manages to launch <i>and</i>
		 * fail a job execution for this instance between the last assertion and the
		 * next method returning successfully.
		 */
		jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);

		try {
			taskScheduler.schedule(new Runnable() {

				@Override
				public void run() {
					try {
						logger.info(
								"Job: [" + job + "] launched with the following parameters: [" + jobParameters + "]");
						job.execute(jobExecution);
						Duration jobExecutionDuration = BatchMetrics.calculateDuration(jobExecution.getStartTime(),
								jobExecution.getEndTime());
						logger.info("Job: [" + job + "] completed with the following parameters: [" + jobParameters
								+ "] and the following status: [" + jobExecution.getStatus() + "]"
								+ (jobExecutionDuration == null ? ""
										: " in " + BatchMetrics.formatDuration(jobExecutionDuration)));
					} catch (Throwable t) {
						logger.info(
								"Job: [" + job + "] failed unexpectedly and fatally with the following parameters: ["
										+ jobParameters + "]",
								t);
						rethrow(t);
					}
				}

				private void rethrow(Throwable t) {
					if (t instanceof RuntimeException) {
						throw (RuntimeException) t;
					} else if (t instanceof Error) {
						throw (Error) t;
					}
					throw new IllegalStateException(t);
				}
			}, jobParameters.getDate(BatchConstants.BATCH_LAUNCH_TIME));
		} catch (TaskRejectedException e) {
			jobExecution.upgradeStatus(BatchStatus.FAILED);
			if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
				jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
			}
			jobRepository.update(jobExecution);
		}

		return jobExecution;
	}

}
